package es;

import co.elastic.clients.elasticsearch.ElasticsearchAsyncClient;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.mapping.Property;
import co.elastic.clients.elasticsearch.core.*;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import co.elastic.clients.elasticsearch.core.bulk.CreateOperation;
import co.elastic.clients.elasticsearch.indices.CreateIndexResponse;
import co.elastic.clients.elasticsearch.indices.DeleteIndexResponse;
import co.elastic.clients.elasticsearch.indices.ElasticsearchIndicesClient;
import co.elastic.clients.elasticsearch.indices.GetIndexResponse;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.ElasticsearchTransport;
import co.elastic.clients.transport.endpoints.BooleanResponse;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import co.elastic.clients.util.ObjectBuilder;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Function;

/**
 * es 入门<p/>
 * <a href="https://www.cnblogs.com/wuyongyin/p/16399197.html">elasticsearch-java API</a><p>
 *
 * @author Verlet
 * @date 2023-03-12 00:24
 */
@Slf4j
public class ElasticsearchDemo {

    private ElasticsearchClient client;

    private ElasticsearchAsyncClient asyncClient;

    private ElasticsearchTransport transport;

    private static final String INDEX_NAME = "user";


    public ElasticsearchDemo() {
        connect();
    }

    /**
     * 创建连接
     */
    public void connect() {
        RestClient restClient = RestClient.builder(new HttpHost("192.168.31.64", 9200)).build();
        ObjectMapper objectMapper = new ObjectMapper();
        transport = new RestClientTransport(restClient, new JacksonJsonpMapper(objectMapper));
        client = new ElasticsearchClient(transport);
        asyncClient = new ElasticsearchAsyncClient(transport);
    }

    /**
     * 关闭连接
     */
    public void close() throws IOException {
        if (transport != null) {
            transport.close();
        }
    }

    /**
     * 创建索引
     */
    public void createIndex() throws IOException {

        Function<Property.Builder, ObjectBuilder<Property>> propertiesText = propertiesBuilder ->
                propertiesBuilder.text(textPropertyBuilder ->
                        textPropertyBuilder.analyzer("ik_max_word").searchAnalyzer("ik_max_word"));

        Function<Property.Builder, ObjectBuilder<Property>> propertiesKeyword = propertiesBuilder ->
                propertiesBuilder.keyword(keywordPropertyBuilder -> keywordPropertyBuilder);

        ElasticsearchIndicesClient indices = client.indices();

        BooleanResponse exists = indices.exists(builder -> builder.index(INDEX_NAME));
        if (exists.value()) {
            log.info("索引:{} 已经存在了", INDEX_NAME);
        } else {
            CreateIndexResponse createIndexResponse = indices.create(builder ->
                    builder.settings(indexSettingsBuilder -> indexSettingsBuilder.numberOfReplicas("1").numberOfShards("2"))
                            .mappings(typeMappingsBuilder -> typeMappingsBuilder
                                    .properties("age", propertiesBuilder ->
                                            propertiesBuilder.integer(integerNumberPropertyBuilder -> integerNumberPropertyBuilder))
                                    .properties("name", propertiesKeyword)
                                    .properties("poems", propertiesText)
                                    .properties("about", propertiesText)
                                    .properties("success", propertiesKeyword)
                            ).index(INDEX_NAME));
            log.info("创建索引：createIndexResponse = {}", createIndexResponse);
        }
    }

    public void getIndex() throws IOException {
        GetIndexResponse getIndexResponse = client.indices().get(builder -> builder.index(INDEX_NAME));
        log.info("获取索引:{}", getIndexResponse.toString());
    }

    /**
     * 插入文档数据
     */
    public void addDocument() throws IOException {
        User user = new User();
        user.setId(1001);
        user.setAge(18);
        user.setName("张三");
        user.setPoems("poems");
        user.setAbout("这是一个我，不一样的我。");
        user.setSuccess("success");

        CreateResponse createResponse = client.create(builder -> builder.index(INDEX_NAME).id("1001").document(user));

        log.info("文档对象创建的响应对象:{}", createResponse);


    }

    /**
     * 批量插入数据
     */
    public void addBulkDocument() throws IOException {
        List<User> users = new ArrayList<>();
        for (int i = 1; i < 6; i++) {
            users.add(new User(2000 + i, "李四" + i, 18, "poems", "我是李四" + i, "success"));
        }

        BulkResponse bulkResponse = client.bulk(builder -> {
                    users.forEach(user ->
                            builder.operations(bulkBuilder ->
                                    bulkBuilder.create(createBuilder ->
                                            createBuilder.index(INDEX_NAME).id(String.valueOf(user.getId())).document(user)
                                    )
                            )
                    );
                    return builder;
                }
        );
        log.info("批量新增数据的响应:{}", bulkResponse);
    }


    /**
     * 删除文档 根据id
     */
    public void deleteDocument() throws IOException {
        DeleteResponse deleteResponse = client.delete(builder -> builder.index(INDEX_NAME).id("1001"));
        log.info("删除文档对象：{}",deleteResponse);
    }

    /**
     * 查询
     */
    public void findDocument() throws IOException {
        SearchResponse<User> search = client.search(builder -> {
            builder.index(INDEX_NAME).query(matchBuilder -> matchBuilder.match(matchQuery -> matchQuery.field("name").query("张三")));
            return builder;
        }, User.class);
        log.info("查询到数据：{}",search.hits().hits().get(0).source());
    }

    /**
     * 异步操作
     */
    public void asyncClientOperation(){
        asyncClient.indices().create(builder -> builder.index("asyncIndex"))
                .thenApply(CreateIndexResponse::acknowledged)
                .whenComplete((resp,error)->{
                    log.info("回调方法");
                    if(resp){
                        log.info("异步索引创建：{}",resp);
                    }else {
                        error.printStackTrace();
                    }
                });
    }


    /**
     * 删除索引
     */
    public void deleteIndex() throws IOException {
        DeleteIndexResponse deleteIndexResponse = client.indices().delete(builder -> builder.index(INDEX_NAME));
        log.info("删除索引 deleteIndexResponse={}", deleteIndexResponse.acknowledged());
    }


    public static void main(String[] args) throws IOException {
        ElasticsearchDemo elasticsearchDemo = new ElasticsearchDemo();

        try {

            elasticsearchDemo.createIndex();

//            elasticsearchDemo.getIndex();
//            elasticsearchDemo.addDocument();
//            elasticsearchDemo.addBulkDocument();
//            elasticsearchDemo.deleteDocument();
//            elasticsearchDemo.findDocument();
            elasticsearchDemo.asyncClientOperation();
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        } finally {
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
//            elasticsearchDemo.deleteIndex();
            elasticsearchDemo.close();
        }

    }


}
